package heap;

import exceptions.DBInvalidSlotNumberException;
import Transaction.Transaction;
import Common.Convert;
import Common.GlobalConst;
import storage.data.Page;
import storage.data.PageID;


/**
 * Manages the placement and access of records on a disk block.
 */

/**
 * Constructor of the slotted-page: Number of the tuples|The ptr of every
 * tuple|Free space|records
 */

interface ConstSlot {
	int INVALID_SLOT = -1;

	int EMPTY_SLOT = -1;
}

public class HFPage extends Page implements GlobalConst, ConstSlot {

	/**
	 * Number of the slots.
	 */
	private short slotCount;

	/**
	 * offset of first used byte by data records in data[]
	 */
	private short usedPtr;

	/**
	 * an arbitrary value used by subclasses as needed
	 */
	private short type;

	/**
	 * number of bytes free in data[]
	 */
	private short freeSpace;

	/**
	 * backward pointer to data page
	 */
	private PageID prevPage = new PageID(-1);

	/**
	 * forward pointer to data page
	 */
	private PageID nextPage = new PageID(-1);

	/**
	 * page number of this page
	 */
	public PageID curPage = new PageID(-1);

	public HFPage() {

	}

	public HFPage(Page page) {
		data = page.getData();

	}

	// initialize a new page
	public void init(PageID pageNo, Page apage, Transaction tx, String table)
			throws Exception {
		data = apage.getData();
		slotCount = 0;
		Convert.setShortValue(slotCount, SLOT_CNT, data, tx, table, pageNo);
		curPage.pageId = pageNo.pageId;
		Convert
				.setIntValue(curPage.pageId, CUR_HFPAGE, data, tx, table,
						pageNo);

		nextPage.pageId = prevPage.pageId = INVALID;
		Convert.setIntValue(prevPage.pageId, PREV_HFPAGE, data, tx, table,
				pageNo);
		Convert.setIntValue(nextPage.pageId, NEXT_HFPAGE, data, tx, table,
				pageNo);

		usedPtr = (short) MAX_FRAME; // offset in data array (grow backwards)
		Convert.setShortValue(usedPtr, USED_PTR, data, tx, table, pageNo);

		freeSpace = (short) (MAX_FRAME - DPFIXED); // amount of space available
		Convert.setShortValue(freeSpace, FREE_SPACE, data, tx, table, pageNo);

	}

	/**
	 * 
	 * @return
	 */
	public byte[] getHFPageData() {
		return data;
	}

	/**
	 * Set and get the parameter or the page.
	 * 
	 * @return
	 * @throws IOException
	 */

	public PageID getPrevPage(Transaction tx, String table) throws Exception {
		prevPage.pageId = Convert.getIntValue(PREV_HFPAGE, data, tx, table);
		return prevPage;
	}

	public void setPrevPage(PageID pageNo, Transaction tx, String table)
			throws Exception {
		prevPage.pageId = pageNo.pageId;
		Convert.setIntValue(prevPage.pageId, PREV_HFPAGE, data, tx, table,
				curPage);
	}

	public PageID getNextPage(Transaction tx, String table) throws Exception {
		nextPage.pageId = Convert.getIntValue(NEXT_HFPAGE, data, tx, table);
		return nextPage;
	}

	public void setNextPage(PageID pageNo, Transaction tx, String table)
			throws Exception {
		nextPage.pageId = pageNo.pageId;
		Convert.setIntValue(nextPage.pageId, NEXT_HFPAGE, data, tx, table,
				curPage);
	}

	public PageID getCurPage(Transaction tx, String table) throws Exception {
		curPage.pageId = Convert.getIntValue(CUR_HFPAGE, data, tx, table);
		return curPage;
	}

	public void setCurPage(PageID pageNo, Transaction tx, String table)
			throws Exception {
		curPage.pageId = pageNo.pageId;
		Convert.setIntValue(curPage.pageId, CUR_HFPAGE, data, tx, table,
				curPage);
	}

	public short getSlotCnt(Transaction tx, String table) throws Exception {
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);
		return slotCount;
	}

	/**
	 * sets slot contents
	 * 
	 * @param slotno
	 *            the slot number
	 * @param length
	 *            length of record the slot contains
	 * @param offset
	 *            offset of record
	 * @exception IOException
	 *                I/O errors
	 */

	public void setSlot(int slotno, int length, int offset, Transaction tx,
			String table, PageID pageNo) throws Exception {
		int position = DPFIXED + slotno * SIZE_OF_SLOT;
		Convert.setShortValue((short) length, position, data, tx, table,
						pageNo);
		Convert.setShortValue((short) offset, position + 2, data, tx, table,
				pageNo);
	}

	/**
	 * @param slotno
	 *            slot number
	 * @exception IOException
	 *                I/O errors
	 * @return the length of record the given slot contains
	 */
	public short getSlotLength(int slotno, Transaction tx, String table)
			throws Exception {
		int position = DPFIXED + slotno * SIZE_OF_SLOT;
		short val = Convert.getShortValue(position, data, tx, table);
		return val;
	}

	/**
	 * @param slotno
	 *            slot number
	 * @exception IOException
	 *                I/O errors
	 * @return the offset of record the given slot contains
	 */
	public short getSlotOffset(int slotno, Transaction tx, String table)
			throws Exception {
		int position = DPFIXED + slotno * SIZE_OF_SLOT;
		short val = Convert.getShortValue(position + 2, data, tx, table);
		return val;
	}

	public short getType(Transaction tx, String table) throws Exception {
		type = Convert.getShortValue(TYPE, data, tx, table);
		return type;
	}

	public void setType(short valtype, Transaction tx, String table,
			PageID pageNo) throws Exception {
		type = valtype;
		Convert.setShortValue(type, TYPE, data, tx, table, pageNo);
	}

	/**
	 * inserts a new record onto the page, returns RID of this record
	 * 
	 * @param record
	 * @return
	 * @throws IOException
	 */
	public RID insertRecord(byte[] record, Transaction tx, String table,
			PageID pageNo) throws Exception {
		RID rid = new RID();

		int recLen = record.length;
		int spaceNeeded = recLen + SIZE_OF_SLOT;
		freeSpace = Convert.getShortValue(FREE_SPACE, data, tx, table);

		if (spaceNeeded > freeSpace) {
			return null;
		} else {
			slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);
//			int num = 0 ;
//			short length;
//			Integer temp = Global.slotInfo.get(curPage.pageId);
//			if ( temp == null ) temp = 0 ;
			//temp = 0 ;
//			int i ;
//			for ( i = 0 ; i < ( slotCount + 1 ); i ++ ) {
//				num = ( temp + i ) %  ( slotCount + 1) ; 
//				//if( i == 0 ) System.out.println ( num ) ; 
//				length = getSlotLength(num, tx, table);
//				if (length == EMPTY_SLOT) break ;
//			}
//
//			System.out.printf  ("page:%d slots:%d i:%d num:%d\n", curPage.pageId, slotCount , i,num ) ;
//			length = getSlotLength(num, tx, table);
//			if ( i < slotCount && length != EMPTY_SLOT) 
//				System.out.println ( "dsfasdfasf") ; 

			
//			int tempnum = 0;
//			if (temp != null) {
//				num = temp % (slotCount + 1);
//				tempnum = num;
//			} else {
//				num = 0;
//			}
//			if (num != slotCount||getSlotLength(num, tx, table)!=EMPTY_SLOT) {
//				for (; num < slotCount; num++) {
//					length = getSlotLength(num, tx, table);
//					if (length == EMPTY_SLOT) {
//						break;
//					}
//				}
//				if(num==slotCount && slotCount!=0) {
//					for(num=0;num<tempnum;num++) {
//						length = getSlotLength(num, tx, table);
//						if (length == EMPTY_SLOT) {
//							break;
//						}
//					}
//					num = slotCount;
//				}
//			}

			int num;
			int length;
			
			for (num = 0; num < slotCount; num++) {
				length = getSlotLength(num,tx,table);
				if (length == EMPTY_SLOT) {
					break;
				}
			}
			if (num == slotCount) {
//			if ( i== slotCount) {
				 //System.out.println("haha");
				freeSpace -= spaceNeeded;
//				num = slotCount ; 
				Convert.setShortValue(freeSpace, FREE_SPACE, data, tx, table,
						pageNo);
				slotCount++;
				Convert.setShortValue(slotCount, SLOT_CNT, data, tx, table,
						pageNo);
			} else {
				 //System.out.println("haha");
				freeSpace -= recLen;
				Convert.setShortValue(freeSpace, FREE_SPACE, data, tx, table,
						pageNo);
			}
			//int pos = num+1  ;
			
			
			//Global.slotInfo.put(curPage.pageId,pos);

			usedPtr = Convert.getShortValue(USED_PTR, data, tx, table);
			usedPtr -= recLen; // adjust usedPtr
			Convert.setShortValue(usedPtr, USED_PTR, data, tx, table, pageNo);
			// insert the slot info onto the data page
			setSlot(num, recLen, usedPtr, tx, table, pageNo);

			// insert data onto the data page
			tx.setBytes(table, curPage, record, 0, data, (usedPtr), recLen);
			curPage.pageId = Convert.getIntValue(CUR_HFPAGE, data, tx, table);
			rid.pageNo.pageId = curPage.pageId;
			rid.slotNo = num;
			return rid;
		}
	}

	public int deleteRecord(RID rid, Transaction tx, String table, PageID pageNo)
			throws Exception, DBInvalidSlotNumberException {
		// System.out.println("PageNo:"+pageNo.pageId+" "+"Slot:"+slotCount);
		int slotNo = rid.slotNo;
		short recLen = getSlotLength(slotNo, tx, table);
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);
		if ((slotNo >= 0) && (slotNo < slotCount) && (recLen > 0)) {
			int offset = getSlotOffset(slotNo, tx, table);
			usedPtr = Convert.getShortValue(USED_PTR, data, tx, table);
			int newSpot = usedPtr + recLen;
			int size = offset - usedPtr;
			// shift bytes to the right
			// System.arraycopy(data, usedPtr, data, newSpot, size);
			tx.setBytes(table, pageNo, data, usedPtr, data, newSpot, size);
			// adjust the offsets of these slots which be moved
			int i, n, chkoffset;
			for (i = 0, n = DPFIXED; i < slotCount; n += SIZE_OF_SLOT, i++) {
				if ((getSlotLength(i, tx, table) >= 0)) {
					chkoffset = getSlotOffset(i, tx, table);
					if (chkoffset < offset) {
						chkoffset += recLen;
						Convert.setShortValue((short) chkoffset, n + 2, data,
								tx, table, pageNo);
					}
				}
			}
			usedPtr += recLen;
			Convert.setShortValue(usedPtr, USED_PTR, data, tx, table, pageNo);
			freeSpace = Convert.getShortValue(FREE_SPACE, data, tx, table);
			freeSpace += recLen;
			Convert.setShortValue(freeSpace, FREE_SPACE, data, tx, table,
					pageNo);

			setSlot(slotNo, EMPTY_SLOT, 0, tx, table, pageNo); // mark slot
																// free
//			Integer tmp ;
//			if((tmp=Global.slotInfo.get(curPage.pageId))!=null) {
//				Global.slotInfo.put ( curPage.pageId, ( tmp - 1 + slotCount ) %  slotCount  ) ;
//			}
			return slotNo;
		} else {
			throw new DBInvalidSlotNumberException(null);
		}
	}

	public RID firstRecord(Transaction tx, String table) throws Exception {
		RID rid = new RID();
		// find the first non-empty slot
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);
		int i;
		short length;
		for (i = 0; i < slotCount; i++) {
			length = getSlotLength(i, tx, table);
			if (length != EMPTY_SLOT)
				break;
		}
		if (i == slotCount)
			return null;
		// found a non-empty slot
		rid.slotNo = i;
		curPage.pageId = Convert.getIntValue(CUR_HFPAGE, data, tx, table);
		rid.pageNo.pageId = curPage.pageId;
		return rid;
	}

	public RID nextRecord(RID curRid, Transaction tx, String table)
			throws Exception {
		RID rid = new RID();
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);

		int i = curRid.slotNo;
		short length;

		// find the next non-empty slot
		for (i++; i < slotCount; i++) {
			length = getSlotLength(i, tx, table);
			if (length != EMPTY_SLOT)
				break;
		}

		if (i >= slotCount)
			return null;

		// found a non-empty slot

		rid.slotNo = i;
		curPage.pageId = Convert.getIntValue(CUR_HFPAGE, data, tx, table);
		rid.pageNo.pageId = curPage.pageId;
		return rid;
	}

	public Tuple getRecord(RID rid, Transaction tx, String table)
			throws Exception, DBInvalidSlotNumberException {
		short recLen;
		short offset;
		byte[] record;
		PageID pageNo = new PageID();
		pageNo.pageId = rid.pageNo.pageId;
		curPage.pageId = Convert.getIntValue(CUR_HFPAGE, data, tx, table);
		int slotNo = rid.slotNo;

		// length of record being returned
		recLen = getSlotLength(slotNo, tx, table);
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);
		if ((slotNo >= 0) && (slotNo < slotCount) && (recLen > 0)
				&& (pageNo.pageId == curPage.pageId)) {
			offset = getSlotOffset(slotNo, tx, table);
			record = new byte[recLen];
			tx.getBytes(table, data, offset, record, 0, recLen);

			Tuple tuple = new Tuple(record, 0, recLen);
			// System.out.println("hahaha:"+tuple.noOfFlds());
			return tuple;
		}

		else {
			throw new DBInvalidSlotNumberException(null);
		}
	}

	public Tuple returnRecord(RID rid, Transaction tx, String table)
			throws Exception, DBInvalidSlotNumberException {
		short recLen;
		short offset;
		PageID pageNo = new PageID();
		pageNo.pageId = rid.pageNo.pageId;

		curPage.pageId = Convert.getIntValue(CUR_HFPAGE, data, tx, table);
		int slotNo = rid.slotNo;

		// length of record being returned
		recLen = getSlotLength(slotNo, tx, table);
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);

		if ((slotNo >= 0) && (slotNo < slotCount) && (recLen > 0)
				&& (pageNo.pageId == curPage.pageId)) {

			offset = getSlotOffset(slotNo, tx, table);
			Tuple tuple = new Tuple(data, offset, recLen);
			return tuple;
		}

		else {
			throw new DBInvalidSlotNumberException(null);
		}

	}

	public int availableSpace(Transaction tx, String table) throws Exception {
		freeSpace = Convert.getShortValue(FREE_SPACE, data, tx, table);
		return (freeSpace - SIZE_OF_SLOT);
	}

	public boolean empty(Transaction tx, String table) throws Exception {
		int i;
		short length;
		// look for an empty slot
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);

		for (i = 0; i < slotCount; i++) {
			length = getSlotLength(i, tx, table);
			if (length != EMPTY_SLOT)
				return false;
		}

		return true;
	}

	/**
	 * Compacts the slot directory on an HFPage.
	 */

	/**
	 * Compacts the slot directory on an HFPage.
	 */

	protected void compactSlotDir(Transaction tx, String table, PageID pageNo,
			int slotNo) throws Exception {
		boolean move = false; // Move a record? -- initially false

//		Global.timer.start("hahahaha");
		slotCount = Convert.getShortValue(SLOT_CNT, data, tx, table);
		freeSpace = Convert.getShortValue(FREE_SPACE, data, tx, table);
		// length = getSlotLength(slotNo,tx,table);
		int pos1 = DPFIXED + (slotNo + 1) * SIZE_OF_SLOT;
		int pos2 = DPFIXED + (slotNo) * SIZE_OF_SLOT;
		int len = (slotCount - slotNo) * SIZE_OF_SLOT;
		move = true;
		tx.setBytes(table, pageNo, data, pos1, data, pos2, len);
		// System.out.println("slotCount:"+slotCount);

		// while (current_scan_posn < slotCount) {
		//			
		// length = getSlotLength(current_scan_posn,tx,table);
		//
		// if ((length == EMPTY_SLOT) && (move == false)) {
		// move = true;
		// first_free_slot = current_scan_posn;
		// } else if ((length != EMPTY_SLOT) && (move == true)) {
		// offset = getSlotOffset(current_scan_posn,tx,table);
		//				
		// // slot[first_free_slot].length =
		// // slot[current_scan_posn].length;
		// // slot[first_free_slot].offset =
		// // slot[current_scan_posn].offset;
		// setSlot(first_free_slot, length, offset,tx,table,pageNo);
		//
		// // Mark the current_scan_posn as empty
		// // slot[current_scan_posn].length = EMPTY_SLOT;
		// setSlot(current_scan_posn, EMPTY_SLOT, 0,tx,table,pageNo);
		//
		// // Now make the first_free_slot point to the next free slot.
		// first_free_slot++;
		//
		// // slot[current_scan_posn].length == EMPTY_SLOT !!
		// while (getSlotLength(first_free_slot,tx,table) != EMPTY_SLOT) {
		// first_free_slot++;
		// }
		// }
		//
		// current_scan_posn++;
		// }
//		Global.timer.end("hahahaha");
		// System.out.println("freeSpace:"+freeSpace);
		if (move == true) {
			// Adjust amount of free space on page and slotCnt
			// freeSpace += SIZE_OF_SLOT * (slotCount - first_free_slot);

			// slotCount = (short) first_free_slot;
			freeSpace += SIZE_OF_SLOT;
			// System.out.println("newfreeSpace:"+freeSpace);
			slotCount -= 1;
			Convert.setShortValue(freeSpace, FREE_SPACE, data, tx, table,
					pageNo);
			Convert.setShortValue(slotCount, SLOT_CNT, data, tx, table, pageNo);
		}
	}
}
